package zhaotupian;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.File;
import java.io.IOException;

public class YeMianTIQu2 {
    //提取一个网页里的图片，例如<img src="http://t1.27270.com/uploads/tu/201812/83/667f2957af.jpg" width="190" height="280" 把网址取出来
//此程序不需要reduce部分

        public static class MapYeMianTiQu extends Mapper<LongWritable, Text, Text, NullWritable> {
            @Override
            protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
                String s = value.toString();
                //string的包含（continue）
                if (s.contains("img src=")){
                    //进行切分
                    String[] line = s.split("img src=\"");//   \" 是转义操作
                    for (String s1 : line) {
                        //判断是否有.jpg的文件，如果有再进行切分
                        if (s1.contains(".jpg")){
                            //取数组的第0位，也就是前面的网址
                            String s2 = s1.split("\" width")[0];
                            context.write(new Text(s2),NullWritable.get());
                        }else {
                            continue;//continue只能在for循环里使用，而return在if和for里都能用，所以这里也可写return
                        }
                    }
                }else {
                    return;
                }
            }

        }

        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            Configuration conf =new Configuration();
            Job job = new Job();

            job.setJarByClass(YeMianTIQu2.class);
            job.setMapperClass(MapYeMianTiQu.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(NullWritable.class);


            File file = new File("f:/all/a/tpc");
            if (file.exists()) {
                FileUtil.fullyDelete(file);
            }
            FileInputFormat.addInputPath(job,new Path("f:/all/a/tp/c.txt"));
            FileOutputFormat.setOutputPath(job,new Path("f:/all/a/tpc"));

            job.setNumReduceTasks(1);
            System.exit(job.waitForCompletion(true)?0:1);
        }
    }
